package com.org.oracle.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.org.oracle.constant.RedisKeyConstants;
import com.org.oracle.mapper.NoteCountMapper;
import com.org.oracle.mapper.UserCountMapper;
import com.org.oracle.mysql.entity.NoteCount;
import com.org.oracle.mysql.entity.UserCount;
import com.org.oracle.mysql.vo.*;
import com.org.oracle.service.CountService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2024/12/11 下午6:26
 * @version: v1.0.0
 * @description: 计数服务
 */
@Slf4j
@Service
public class CountServiceImpl implements CountService {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource(name = "countExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private UserCountMapper userCountMapper;

    @Resource
    private NoteCountMapper noteCountMapper;

    /**
     * 根据用户 ID 查询用户的计数信息
     *
     * @param uid
     * @return
     */
    @Override
    public UserCountVO findUserCount(String uid) {
        // 先在 Redis 查询相关用户计数信息
        String countUserKey = RedisKeyConstants.buildCountUserKey(uid);
        Map<Object, Object> userCountMap = redisTemplate.opsForHash().entries(countUserKey);

        // Redis 中存在用户的计数信息
        if (CollUtil.isNotEmpty(userCountMap)) {
            // 转化成 Map<String, Long> 类型
            Map<String, Long> map = userCountMap.entrySet().stream().collect(Collectors.toMap(
                    key -> key.getKey().toString(),
                    value -> Long.valueOf(value.getValue().toString())
            ));

            // 构建反参
            UserCountVO userCountVO = new UserCountVO();

            // 反参赋值
            map.forEach((k, v) -> {
                // 根据 key 来进行赋值
                switch (k) {
                    case RedisKeyConstants.FIELD_LIKE_TOTAL -> userCountVO.setLikeTotal(v);
                    case RedisKeyConstants.FIELD_NOTE_TOTAL -> userCountVO.setNoteTotal(v);
                    case RedisKeyConstants.FIELD_BRIGHT_TOTAL -> userCountVO.setBrightTotal(v);
                }
            });

            return userCountVO;
        }

        // 当 userCountMap 不为空时说明用户计数 Redis 没初始化，需要从数据库中查询数据后，初始化 Redis
        // 从数据库中查询用户计数信息
        LambdaQueryWrapper<UserCount> lambdaQueryWrapper = Wrappers.<UserCount>lambdaQuery()
                .select(UserCount::getUid, UserCount::getNoteTotal, UserCount::getLikeTotal, UserCount::getBrightTotal)
                .eq(UserCount::getUid, uid);
        UserCount userCount = userCountMapper.selectOne(lambdaQueryWrapper);

        // 这里存在一种特殊情况，因为当用户没收到点赞或点亮时不会在数据库创建任何计数数据
        // 因此就算计数数据为空也要返回对应的信息并上传到 Redis 中
        // 用户计数数据不存在，缓存设为空值
        if (Objects.isNull(userCount)) {
            UserCount userCountNotExist = UserCount.builder()
                    .noteTotal(0L)
                    .likeTotal(0L)
                    .brightTotal(0L)
                    .build();
            // 异步写入 Redis
            threadPoolTaskExecutor.submit(() -> batchAddUserCountHashAndExpire(userCountNotExist, countUserKey));
            return UserCountVO.builder()
                    .uid(uid)
                    .noteTotal(0L)
                    .likeTotal(0L)
                    .brightTotal(0L)
                    .build();
        }

        // 异步写入 Redis
        threadPoolTaskExecutor.submit(() -> batchAddUserCountHashAndExpire(userCount, countUserKey));

        return UserCountVO.builder()
                .uid(uid)
                .likeTotal(userCount.getLikeTotal())
                .noteTotal(userCount.getNoteTotal())
                .brightTotal(userCount.getBrightTotal())
                .build();
    }

    /**
     * 批量查询笔记被点赞的数量
     *
     * @param ids
     * @return
     */
    @Override
    public List<NoteLikeTotalVO> findNoteLikeTotalByNoteIds(List<String> ids) {
        // 先从 redis 中查询笔记的计数信息是否存在
        Map<String, String> redisKeys = ids.stream().collect(Collectors.toMap(RedisKeyConstants::buildCountNoteKey, id -> id));

        // 筛选出缓存里没有的笔记计数数据，去查数据库
        List<String> noteIdsNeedQuery = Lists.newArrayList();
        // 构建反参
        List<NoteLikeTotalVO> noteLikeTotalVOS = Lists.newArrayList();

        // 使用 redis 管道批量查询
        redisTemplate.executePipelined((RedisCallback<?>) connection -> {
            for (Map.Entry<String, String> entry : redisKeys.entrySet()) {
                String key = entry.getKey();
                String value = entry.getValue();
                // 获取到点赞数
                Object likeTotalRedisValue = redisTemplate.opsForHash().get(key, RedisKeyConstants.FIELD_LIKE_TOTAL);

                // 判断不为空
                if (Objects.nonNull(likeTotalRedisValue)) {
                    long likeTotal = Long.parseLong(likeTotalRedisValue.toString());
                    // 添加 noteLikeTotalVO
                    NoteLikeTotalVO noteLikeTotalVO = NoteLikeTotalVO.builder()
                            .noteId(value)
                            .likeTotal(likeTotal)
                            .build();
                    noteLikeTotalVOS.add(noteLikeTotalVO);
                    continue;
                }
                // 记录下不存在 redis 的数据
                noteIdsNeedQuery.add(value);
                // 先设置为 0 防止缓存穿透
                redisTemplate.opsForHash().put(key, RedisKeyConstants.FIELD_LIKE_TOTAL, 0);
                // 保底 1 分钟 + 随机秒数
                long expireSeconds = 60 + RandomUtil.randomInt(60);
                redisTemplate.expire(key, expireSeconds, TimeUnit.SECONDS);
            }
            return null;
        });

        // 如果所有数据已在缓存中，直接返回
        if (CollUtil.isEmpty(noteIdsNeedQuery)) {
            return noteLikeTotalVOS;
        }

        // redis 数据可能存在一部分也可能全部都不存在，查询数据库获取数据
        // 根据笔记 Ids 查询笔记的获得点赞计数
        LambdaQueryWrapper<NoteCount> lambdaQueryWrapper = Wrappers.<NoteCount>lambdaQuery()
                .select(NoteCount::getNoteId,
                        NoteCount::getLikeTotal,
                        NoteCount::getCollectTotal,
                        NoteCount::getCommentTotal)
                .in(NoteCount::getNoteId, noteIdsNeedQuery);
        List<NoteCount> noteCounts = noteCountMapper.selectList(lambdaQueryWrapper);

        // 这里存在一种特殊情况，因为当笔记没点赞或收藏时不会在数据库创建任何计数数据
        // 因此就算有些计数数据为空也要返回对应的信息并上传到 Redis 中
        Map<String, NoteCount> noteCountMap = noteCounts.stream()
                .collect(Collectors.toMap(NoteCount::getNoteId, value -> value));

        // 将不存在数据库的笔记 ID 构建出 NoteCount 上传到 Redis 中
        List<NoteCount> noteExistNoteCounts = noteIdsNeedQuery.stream()
                .filter(noteId -> !noteCountMap.containsKey(noteId))
                .map(noteId -> NoteCount.builder()
                        .noteId(noteId)
                        .likeTotal(0L)
                        .collectTotal(0L)
                        .commentTotal(0L)
                        .build())
                .toList();

        // 合并需要同步 Redis 的数据
        noteCounts.addAll(noteExistNoteCounts);

        // 修改为反参类型
        List<NoteLikeTotalVO> noteLikeTotalVOS2 = noteCounts.stream()
                .map(noteCount -> NoteLikeTotalVO.builder()
                        .noteId(noteCount.getNoteId())
                        .likeTotal(noteCount.getLikeTotal())
                        .build())
                .toList();

        // 数据合并
        noteLikeTotalVOS.addAll(noteLikeTotalVOS2);

        // 将数据库数据异步同步到 redis 中
        threadPoolTaskExecutor.submit(() -> {
            // 准备进行 redis 管道传递的参数
            Map<String, Map<String, Long>> executePipelinedArgs = Maps.newHashMap();
            noteCounts.forEach(noteCount -> {
                // 构建 redisKey
                String countNoteKey = RedisKeyConstants.buildCountNoteKey(noteCount.getNoteId());
                // 添加到管道参数中
                executePipelinedArgs.put(countNoteKey, Map.of(
                        RedisKeyConstants.FIELD_LIKE_TOTAL, noteCount.getLikeTotal(),
                        RedisKeyConstants.FIELD_COLLECT_TOTAL, noteCount.getCollectTotal(),
                        RedisKeyConstants.FIELD_COMMENT_TOTAL, noteCount.getCommentTotal()
                ));
            });

            // 管道批量添加
            redisTemplate.executePipelined((RedisCallback<?>) connection -> {
                executePipelinedArgs.forEach((k, v) -> {
                    // 将 hashMap 添加到 redisKey 中
                    redisTemplate.opsForHash().putAll(k, v);
                    // 保底 1 天 + 随机秒数
                    long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);
                    // 添加过期时间
                    redisTemplate.expire(k, expireSeconds, TimeUnit.SECONDS);
                });
                return null;
            });
        });

        return noteLikeTotalVOS;
    }

    /**
     * 根据笔记 ID 查询笔记计数信息
     *
     * @param noteId
     * @return
     */
    @Override
    public NoteCountVO findNoteCountByNoteId(String noteId) {
        // 构建 RedisKey
        String countNoteKey = RedisKeyConstants.buildCountNoteKey(noteId);
        Map<Object, Object> noteCountMap = redisTemplate.opsForHash().entries(countNoteKey);

        // Redis 中存在笔记的计数信息
        if (CollUtil.isNotEmpty(noteCountMap)) {
            // 转化类型
            Map<String, Long> map = noteCountMap.entrySet().stream().collect(Collectors.toMap(
                    key -> key.getKey().toString(),
                    value -> Long.valueOf(value.getValue().toString())
            ));

            // 构建反参
            NoteCountVO noteCountVO = NoteCountVO.builder()
                    .noteId(noteId)
                    .build();

            // 反参赋值
            map.forEach((k, v) -> {
                // 根据 key 来进行赋值
                switch (k) {
                    case RedisKeyConstants.FIELD_LIKE_TOTAL -> noteCountVO.setLikeTotal(v);
                    case RedisKeyConstants.FIELD_COLLECT_TOTAL -> noteCountVO.setCollectTotal(v);
                    case RedisKeyConstants.FIELD_COMMENT_TOTAL -> noteCountVO.setCommentTotal(v);
                }
            });

            return noteCountVO;
        }

        // 这里存在一种特殊情况，因为当笔记没收到点赞或收藏时不会在数据库创建任何计数数据
        // 因此就算计数数据为空也要返回对应的信息并上传到 Redis 中
        // 查询数据库
        LambdaQueryWrapper<NoteCount> lambdaQueryWrapper = Wrappers.<NoteCount>lambdaQuery()
                .select(NoteCount::getNoteId, NoteCount::getLikeTotal, NoteCount::getCollectTotal, NoteCount::getCommentTotal)
                .eq(NoteCount::getNoteId, noteId);
        NoteCount noteCount = noteCountMapper.selectOne(lambdaQueryWrapper);

        // 笔记计数数据不存在，设为空值
        if (Objects.isNull(noteCount)) {
            NoteCount noteCountNotExist = NoteCount.builder()
                    .likeTotal(0L)
                    .collectTotal(0L)
                    .commentTotal(0L)
                    .build();
            threadPoolTaskExecutor.submit(() -> batchAddNoteCountHashAndExpire(noteCountNotExist, countNoteKey));

            return NoteCountVO.builder()
                    .noteId(noteId)
                    .likeTotal(0L)
                    .collectTotal(0L)
                    .commentTotal(0L)
                    .build();
        }

        // 异步写入 Redis
        threadPoolTaskExecutor.submit(() -> batchAddNoteCountHashAndExpire(noteCount, countNoteKey));

        return NoteCountVO.builder()
                .noteId(noteId)
                .likeTotal(noteCount.getLikeTotal())
                .collectTotal(noteCount.getCollectTotal())
                .commentTotal(noteCount.getCommentTotal())
                .build();
    }

    /**
     * 批量将笔记计数添加到 Hash 中
     *
     * @param noteCount
     * @param countNoteKey
     */
    private void batchAddNoteCountHashAndExpire(NoteCount noteCount, String countNoteKey) {
        // 保底 1 天 + 随机秒数
        long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);
        // 构建脚本参数
        Object[] args = Lists.newArrayList(
                // 将 key，value 添加到脚本参数中
                RedisKeyConstants.FIELD_LIKE_TOTAL, noteCount.getLikeTotal(),
                RedisKeyConstants.FIELD_COLLECT_TOTAL, noteCount.getCollectTotal(),
                RedisKeyConstants.FIELD_COMMENT_TOTAL, noteCount.getCommentTotal(),
                // 最后一个过期时间
                expireSeconds
        ).toArray();

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/batch_add_note_count_hash_and_expire.lua")));
        // 返回类型
        script.setResultType(Long.class);
        // 执行 Lua 脚本
        redisTemplate.execute(script, Collections.singletonList(countNoteKey), args);
    }

    /**
     * 批量将用户计数添加到 Hash 中
     *
     * @param userCount
     * @param countUserKey
     */
    private void batchAddUserCountHashAndExpire(UserCount userCount, String countUserKey) {
        // 保底 1 天 + 随机秒数
        long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

        // 构建脚本参数
        Object[] args = Lists.newArrayList(
                // 将 key，value 添加到脚本参数中
                RedisKeyConstants.FIELD_LIKE_TOTAL, userCount.getLikeTotal(),
                RedisKeyConstants.FIELD_NOTE_TOTAL, userCount.getNoteTotal(),
                RedisKeyConstants.FIELD_BRIGHT_TOTAL, userCount.getBrightTotal(),
                // 最后一个过期时间
                expireSeconds
        ).toArray();

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/batch_add_user_count_hash_and_expire.lua")));
        // 返回类型
        script.setResultType(Long.class);
        // 执行 Lua 脚本
        redisTemplate.execute(script, Collections.singletonList(countUserKey), args);
    }
}